/**
 * 
 */
package dataflowscheme.engine.core;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * It checks an activity of a data-flow program by checking
 * activity of Event Parser and Execution Event Parser and its 
 * tasks. If an inactivity is recognized and no Stopper received
 * all required tokens then a program is ended. 
 * 
 * @author <a href="mailto:misiak7@gmail.com">Michal Antolik</a>
 *
 */
public class ActivityChecker implements Runnable {

	private Lock lock = new ReentrantLock();
	
	/** set of execution events processed by Execution event parser thread */
	private final Set<Future<?>> peSet = new HashSet<Future<?>>();
	
	/** synchronization aim of Main thread. In program runtime 
	 * phase Main thread is waiting while doneSignal is decreased to zero */
	private final CountDownLatch doneSignal;
	
	/** An interval of checking program scheme */
	private long CHECK_INTERVAL = 1000L; //can be smaller
	
	/** have to be atomic since it is accessed from this and 
	 * event parser threads. It marks an activity of event 
	 * parser, true if parser is active (do processing) */
	private AtomicBoolean epa = new AtomicBoolean(true);
	
	/** is an parser inactivity is recognized then it is set to false
	 * to verify an inactivity in next Activity checker iteration  */
	private boolean lastEPAstate = true;
	
	/**
	 * set to true, if deadlock (an inactivity) was found
	 */
	public boolean wasDeadLock = false;
	
	public ActivityChecker(CountDownLatch doneSignal) {
		this.doneSignal = doneSignal;
	}
	
	public void run(){
		try{
			while(doneSignal.getCount() != 0){ //done signal can be decreased by Stopper
				Thread.sleep(CHECK_INTERVAL);
				
				//duplicate peSet to do not block execution threads while processing events state
				Set<Future<?>> peDuplSet = new HashSet<Future<?>>();
				try{
					lock.lock();
					
					if(peSet.isEmpty() && (epa.get() == false) && (lastEPAstate == false)){
						//verification done
						break;
					}
					
					if(peSet.isEmpty() && (epa.get() == false)){
						//prepare for verification
						lastEPAstate = false;
						continue;
					}
					peDuplSet.addAll(peSet);
				}finally{
					lock.unlock();
				}
				
				
				//finished events will be removed from peSet
				Set<Future<?>> finishedPEs = new HashSet<Future<?>>();
				for(Future<?> pe : peDuplSet){
					if(pe.isDone()){
						finishedPEs.add(pe);
					}
				}
				
				try{
					lock.lock(); //since peSet is accessible from execution event parser thread
					for(Future<?> pe : finishedPEs){
						peSet.remove(pe);
					}
				}finally{
					lock.unlock();
				}
			}
			
			if(doneSignal.getCount() != 0){
				wasDeadLock = true;
				doneSignal.countDown();
			}
		}catch(InterruptedException ie){
			ie.printStackTrace();
		}
	}
	
	/**
	 * add event to observe its state
	 * 
	 * @param pe execution event
	 */
	public void add(Future<?> pe){
		try{
			lock.lock();
			if(pe != null){ //null for end events
				peSet.add(pe);
			}
		}finally{
			lock.unlock();
		}
	}
	
	/**
	 * set activity for event parser
	 */
	public void setEventParserActivity(){
		epa.set(true);
	}
	
	/**
	 * set inactivity for event parser
	 */
	public void setEventParserInactivity(){
		epa.set(false);
	}
	
	

}
